package com.bw.ads;

import com.bw.bean.Mn5TM4Wide;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class AsyncIOFunc4 extends RichAsyncFunction<Mn5TM4Wide,Mn5TM4Wide> {
    Connection conn = null;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/tms01?characterEncoding=utf8", "root", "123456");

        // 机构的名字
        ps1 = conn.prepareStatement("select org_name,region_id from base_organ where id = ?");
        // 省份的名字
        ps2 = conn.prepareStatement("select name from base_province where area_code = ?");
    }

    @Override
    public void close() throws Exception {
        ps2.close();
        ps1.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(Mn5TM4Wide mn5TM4Wide, ResultFuture<Mn5TM4Wide> resultFuture) throws Exception {
        //查询起始站名称 和 区域id 和省份
        ps1.setLong(1, mn5TM4Wide.getStart_org_id());
        ResultSet rs1 = ps1.executeQuery();
        if (rs1.next()) {
            String org_name = rs1.getString("org_name");
            Long region_id = rs1.getLong("region_id");
            mn5TM4Wide.setStart_org_name(org_name);
            // 441234--440000
            Long area_code = Long.valueOf(region_id.toString().substring(0, 2) + "0000");
            ps2.setLong(1, area_code);
            ResultSet rs2 = ps2.executeQuery();
            if (rs2.next()) {
                String name = rs2.getString("name");
                mn5TM4Wide.setStart_province_name(name);
            }
        }

        //结束站相关名称，省份信号箱
        ps1.setLong(1, mn5TM4Wide.getEnd_org_id());
        ResultSet rs3 = ps1.executeQuery();
        if (rs3.next()) {
            String org_name = rs3.getString("org_name");
            Long region_id = rs3.getLong("region_id");
            mn5TM4Wide.setEnd_org_name(org_name);
            Long area_code = Long.valueOf(region_id.toString().substring(0, 2) + "0000");
            ps2.setLong(1, area_code);
            ResultSet rs4 = ps2.executeQuery();
            if (rs4.next()) {
                String name = rs4.getString("name");
                mn5TM4Wide.setEnd_province_name(name);
            }
        }

        //

        resultFuture.complete(Collections.singleton(mn5TM4Wide));
    }
}